#pragma once

#include "muduo/protobuf/codec.h"
#include "muduo/protobuf/dispatcher.h"
#include "muduo/base/Logging.h"
#include "muduo/base/Mutex.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/TcpServer.h"

#include "../mqserver/mq_host.hpp"
#include "../mqserver/mq_connection.hpp"
#include "../mqcommon/mq_threadpool.hpp"

namespace mq
{
    #define HOSTNAME "MyVirtualHost"
    #define DBFILE "/meta.db"
   
    class Server
    {
    private:
     using MessagePtr = std::shared_ptr<google::protobuf::Message>;
        muduo::net::TcpServer _server;
        muduo::net::EventLoop _baseloop;
        ProtobufCodecPtr _codec;        // 协议处理器 对收到的请求进行protobuf协议处理
        ProtobufDispatcher _dispatcher; // 请求分发器
        VirtualHost::ptr _host;
        ConsumerManager::ptr _consumer_manager;
        ConnectionManager::ptr _connection_manager;
        ThreadPool::ptr _pool;

    private:
        void onUnknownMessage(const muduo::net::TcpConnectionPtr &conn,
                              const MessagePtr &message,
                              muduo::Timestamp)
        {
            LOG_INFO << "onUnknownMessage: " << message->GetTypeName();
            conn->shutdown();
        }

        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected() == true)
            {
                LOG_INFO << "新连接建立了";
                _connection_manager->newConnection(_host,_consumer_manager,_codec,conn,_pool);
            }
            else
            {
                LOG_INFO << "连接断开了";
                _connection_manager->delConnection(conn);
            }
        }
    private:
        //打开信道
        void OnOpenChannel(const muduo::net::TcpConnectionPtr &conn,const openChannelRequestPtr &message,muduo::Timestamp){
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr)
            {
                DLOG("打开信道时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }

            return mconn->openChannel(message);
        }
        //关闭信道
        void OnClodeChannle(const muduo::net::TcpConnectionPtr &conn,const closeChannelRequestPtr &message,muduo::Timestamp){
            Connection::ptr mconn = _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr)
            {
                DLOG("关闭信道时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }

            return mconn->closeChannel(message);
        }
        //声明交换机
        void OnDeclareExchange(const muduo::net::TcpConnectionPtr &conn,const declareExchangeRequestPtr &message,muduo::Timestamp){
           Connection::ptr mconn =  _connection_manager->getConnection(conn);
           if(mconn.get() == nullptr)
            {
                DLOG("声明交换机时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }

            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get() == nullptr)
            {
                DLOG("声明交换机时，没有找到信道！");
                conn->shutdown();
                return;
            }

            return cp->declareExchange(message);
        }
        //删除交换机        
        void OnDeleteExchange(const muduo::net::TcpConnectionPtr &conn,const deleteExchangeRequestPtr &message,muduo::Timestamp)
        {
            Connection::ptr mconn =  _connection_manager->getConnection(conn);
           if(mconn.get() == nullptr)
            {
                DLOG("删除交换机时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }

            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get() == nullptr)
            {
                DLOG("删除交换机时，没有找到信道！");
                conn->shutdown();
                return;
            }

            return cp->deleteExchange(message);
        }
        //声明队列
        void OnDeclareQueue(const muduo::net::TcpConnectionPtr &conn,const declareQueueRequestPtr &message,muduo::Timestamp)
        {
            Connection::ptr mconn =  _connection_manager->getConnection(conn);
           if(mconn.get() == nullptr)
            {
                DLOG("声明队列时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }

            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get() == nullptr)
            {
                DLOG("声明队列时，没有找到信道！");
                conn->shutdown();
                return;
            }

            return cp->declareQueue(message);
        }
        //删除队列
         void OnDeleteQueue(const muduo::net::TcpConnectionPtr &conn,const deleteQueueRequestPtr &message,muduo::Timestamp)
         {
            Connection::ptr mconn =  _connection_manager->getConnection(conn);
           if(mconn.get() == nullptr)
            {
                DLOG("删除队列时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }

            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get() == nullptr)
            {
                DLOG("删除队列时，没有找到信道！");
                conn->shutdown();
                return;
            }

            return cp->deleteQueue(message);
         }
        //队列绑定
        void onQueueBind(const muduo::net::TcpConnectionPtr &conn,const queueBindRequestPtr &message,muduo::Timestamp)
        {
            Connection::ptr mconn =  _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr)
            {
                DLOG("绑定队列 时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }

            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get() == nullptr)
            {
                DLOG("绑定队列时，没有找到信道！");
                conn->shutdown();
                return;
            }

            return cp->queueBind(message);
        }
        //队列解绑
        void onQueueUnBind(const muduo::net::TcpConnectionPtr &conn,const queueUnBindRequestPtr &message,muduo::Timestamp)
        {
            Connection::ptr mconn =  _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr)
            {
                DLOG("解除队列绑定时，没有找队列 时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }

            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get() == nullptr)
            {
                DLOG("解除队列绑定时，没有找到信道！");
                conn->shutdown();
                return;
            }

            return cp->queueUnBind(message);
        }
        //消息发布
        void onBasicPublish(const muduo::net::TcpConnectionPtr &conn,const basicPublishRequestPtr &message,muduo::Timestamp)
        {
            Connection::ptr mconn =  _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr)
            {
                DLOG("消息发布时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }

            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get() == nullptr)
            {
                DLOG("消息发布时，没有找到信道！");
                conn->shutdown();
                return;
            }

            return cp->basicPublish(message);
        }
        //消息确认
        void onBasicAck(const muduo::net::TcpConnectionPtr &conn,const basicAckRequestPtr &message,muduo::Timestamp)
        {
            Connection::ptr mconn =  _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr)
            {
                DLOG("消息确认时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }

            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get() == nullptr)
            {
                DLOG("消息确认时，没有找到信道！");
                conn->shutdown();
                return;
            }

            return cp->basicAck(message);
        }
        //订阅队列
        void onBasicConsume(const muduo::net::TcpConnectionPtr& conn,const basicConsumeRequestPtr& message, muduo::Timestamp)
        {
            Connection::ptr mconn =  _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr)
            {
                DLOG("订阅队列时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }

            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get() == nullptr)
            {
                DLOG("订阅队列时，没有找到信道！");
                conn->shutdown();
                return;
            }

            return cp->basicConsume(message);
        }
        //取消订阅
        void onBasicCancel(const muduo::net::TcpConnectionPtr& conn,const basicCancelRequestPtr& message, muduo::Timestamp)
        {
            Connection::ptr mconn =  _connection_manager->getConnection(conn);
            if(mconn.get() == nullptr)
            {
                DLOG("取消订阅时，没有找到连接对应的Connection对象！");
                conn->shutdown();
                return;
            }

            Channel::ptr cp = mconn->getChannel(message->cid());
            if(cp.get() == nullptr)
            {
                DLOG("取消订阅时，没有找到信道！");
                conn->shutdown();
                return;
            }

            return cp->basicCancel(message);
        }

    public:
        Server(int port,const std::string &basedir) : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "server", muduo::net::TcpServer::kReusePort),
                           _codec(std::make_shared<ProtobufCodec>(std::bind(&ProtobufDispatcher::onProtobufMessage, &_dispatcher, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3))),
                           _dispatcher(std::bind(&Server::onUnknownMessage,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3)),
                           _host(std::make_shared<VirtualHost>(HOSTNAME,basedir,basedir + DBFILE)),
                           _consumer_manager(std::make_shared<ConsumerManager>()),
                           _connection_manager(std::make_shared<ConnectionManager>()),
                           _pool(std::make_shared<ThreadPool>())
        {
            //消费者是按照队列为单元进行管理的,针对历史消息中的所有队列，需要初始化队列的消费者管理结构QueueConsumer
            std::unordered_map<std::string, mq::MsgQueue::ptr> mqmp =  _host->allQueue();
            for(auto &mq : mqmp)
            {
                _consumer_manager->initQueueConsumer(mq.first);
            }

            _server.setConnectionCallback(std::bind(&Server::onConnection, this, std::placeholders::_1));
            _server.setMessageCallback(std::bind(&ProtobufCodec::onMessage, _codec.get(), std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            // 注册请求处理函数
            _dispatcher.registerMessageCallback<openChannelRequest>(std::bind(&Server::OnOpenChannel,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<closeChannelRequest>(std::bind(&Server::OnClodeChannle,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<declareExchangeRequest>(std::bind(&Server::OnDeclareExchange,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<deleteExchangeRequest>(std::bind(&Server::OnDeleteExchange,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<declareQueueRequest>(std::bind(&Server::OnDeclareQueue,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<deleteQueueRequest>(std::bind(&Server::OnDeleteQueue,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<queueBindRequest>(std::bind(&Server::onQueueBind,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<queueUnBindRequest>(std::bind(&Server::onQueueUnBind,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicPublishRequest>(std::bind(&Server::onBasicPublish,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicAckRequest>(std::bind(&Server::onBasicAck,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicConsumeRequest>(std::bind(&Server::onBasicConsume,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
            _dispatcher.registerMessageCallback<basicCancelRequest>(std::bind(&Server::onBasicCancel,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3));
        }

        void Start()
        {
            _server.start();
            _baseloop.loop();
        }
    };

};